package cloud.xiguapi.ubas.analysis.market.model;

import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.HashMap;
import java.util.Map;

/**
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-19 上午 11:59
 * desc:
 */
public class SummaryMarketingCountResult implements WindowFunction<Long, SummaryMarketingResult, Tuple, TimeWindow> {

    public static final Map<String, Object> CACHE_MAP = new HashMap<>(16);

    @Override
    public void apply(Tuple tuple, TimeWindow window, Iterable<Long> input, Collector<SummaryMarketingResult> out) throws Exception {
        SummaryMarketingResult result = SummaryMarketingResult.builder()
                .windowEnd(new Timestamp(window.getEnd()).toString())
                .cnt(input.iterator().next())
                .build();

        CACHE_MAP.putIfAbsent(result.getWindowEnd(), result.getCnt());

        out.collect(result);
    }
}